package com.wtw.graph

import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.{SparkConf, SparkContext}

object MapGraphXDemo2 {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName("SimpleGraphX").setMaster("local")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val vertexArray = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42)),
      (5L, ("Ed", 55)),
      (6L, ("Fran", 50))
    )

    val edgeArray = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1),
      Edge(5L, 2L, 2),
      Edge(5L, 3L, 8),
      Edge(5L, 6L, 3)
    )

    //    （1）通过上面的项点数据和边数据创建图对象
    val vertexs = sc.parallelize(vertexArray)
    val edges = sc.parallelize(edgeArray)
    val defaultVertex = (0L, ("Kobe", 1))

    val graph = Graph(vertexs, edges)

    //    （2）对上述的顶点数据和边数据进行修改，再创建一张图2，使得图2中有一些点和边在图1中不存在，
    //     然后调用图1的mask方法，传入图2作为参数，观察返回的图3的信息
    val vertexArray2 = Array(
      (1L, ("Alice", 28)),
      (2L, ("Bob", 27)),
      (3L, ("Charlie", 65)),
      (4L, ("David", 42))
//      (10L, ("James", 10)),
//      (11L, ("Durant", 20)),
//      (12L, ("Curry", 30))
    )

    val edgeArray2 = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 4L, 2),
      Edge(3L, 2L, 4),
//      Edge(3L, 6L, 3),
      Edge(4L, 1L, 1)
//      Edge(5L, 7L, 10),
//      Edge(4L, 7L, 20),
//      Edge(3L, 7L, 30)
    )

    val vertex2 = sc.parallelize(vertexArray2)
    val edges2 = sc.parallelize(edgeArray2)

    val graph2 = Graph(vertex2, edges2)
    val graph3 = graph.mask(graph2)
    graph3.edges.collect().foreach(println(_))
    graph3.vertices.collect().foreach(println(_))
    println("-----------mark-------------")
    //    （3）基于上面的数据进行修改练习subgraph 、groupEdges 、reverses方法

    val re_graph = graph.reverse
    re_graph.edges.collect().foreach(println(_))
    println("-----------reverse-------------")

    val sub_graph = graph.subgraph(epred = (ed) => {ed.srcId != 2})
    sub_graph.edges.collect().foreach(println(_))
    println("-----------subgraph-------------")

    //把起点终点相同的边的属性合并

    val edgeArray3 = Array(
      Edge(2L, 1L, 7),
      Edge(2L, 1L, 2),
      Edge(3L, 6L, 4),
      Edge(3L, 6L, 2)
    )

    val vertex4 = sc.parallelize(vertexArray)
    val edges4 = sc.parallelize(edgeArray3)

    val graph4 = Graph(vertex4, edges4)

    val group_graph: Graph[(String, Int), Int] = graph4.groupEdges(merge = (ed1, ed2) => {ed1 * 10 + ed2})
    group_graph.edges.collect().foreach(println(_))
    println("-----------groupEdges-------------")
  }
}
